package t20240425;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Test {


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);


        String sourceSql = "CREATE TABLE datagen (" +
                "    stime INT," +
                "    sn STRING" +
                "  ) WITH (" +
                "    'connector' = 'datagen'," +
                "    'rows-per-second'='1'," +
                "    'fields.stime.kind'='sequence'," +
                "    'fields.stime.start'='1'," +
                "    'fields.stime.end'='10000'," +
                "    'fields.sn.length'='1'" +
                "  )";

        String sinkSql =
                "  CREATE TABLE print_table (" +
                "    stime INT," +
                "    sn STRING," +
                "    updatetime BIGINT" +
                "    ) WITH (" +
                "    'connector' = 'print'" +
                "  )";

//        String execSql = "INSERT INTO print_table select stime,sn,CAST(CURRENT_TIMESTAMP AS BIGINT) as updatetime from datagen";
        String execSql = "INSERT INTO print_table select stime,sn,UNIX_TIMESTAMP(DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd hh:mm:ss'), 'yyyy-MM-dd hh:mm:ss') as updatetime from datagen";

        tableEnvironment.executeSql(sourceSql);
        System.out.println("-----1");
        tableEnvironment.executeSql(sinkSql);
        System.out.println("-----2");
        tableEnvironment.executeSql(execSql);


    }

}
